
/*
 * Step3HadoopReducer.java
 *
 * Created on May 10, 2012, 1:05:24 AM
 */

package Step3;


import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;
// import org.apache.commons.logging.Log;
// import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;

import BasicTypes.Constants;
import UnionFind.UnionFindHashTable;

/**
 *
 * @author crazymooner
 */
public class Step3HadoopReducer extends MapReduceBase implements Reducer<IntWritable,Text,IntWritable,IntWritable> {
    // The Karmasphere Studio Workflow Log displays logging from Apache Commons Logging, for example:
    // private static final Log LOG = LogFactory.getLog("Step3.Step3HadoopReducer");

	@Override
    public void reduce(IntWritable key, Iterator<Text> value, OutputCollector<IntWritable, IntWritable> output, Reporter reporter)
            throws IOException {
       int group = key.get();
       UnionFindHashTable uf = new UnionFindHashTable();
       //QuickUnion qu = new QuickUnion((Constants.COLUMESIZE + 1) * Constants.SIZE, group);
       while (value.hasNext()) {
    	StringTokenizer st = new StringTokenizer(value.next().toString());
		int p = Integer.parseInt(st.nextToken());
		int q = Integer.parseInt(st.nextToken());
		uf.unite(p, q);
       }
       
      /* for(int i = 0; i < Constants.SIZE; i++){
    	   int index = i + group * Constants.GROUPSIZE;
    	   output.collect(new IntWritable(index), new IntWritable(uf.getRoot(index)));
    	   if(index + Constants.GROUPSIZE < Constants.SIZE * Constants.SIZE){
    	   output.collect(new IntWritable(index + Constants.GROUPSIZE), new IntWritable(uf.getRoot(index + Constants.GROUPSIZE)));
    	   }
       }*/
       for(int i = 0; i < Constants.GROUPSIZE; i++){
    	   int index = i + group * Constants.GROUPSIZE;
    	   output.collect(new IntWritable(index), new IntWritable(uf.getRoot(index)));
       }
    }
}
